import java.io.File;
import java.io.FileNotFoundException;
import java.io.FileReader;
import java.io.LineNumberReader;
import java.util.Map;
import java.util.TreeMap;
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicLong;

/**
 * @author Panbo.Guo
 */
public class Example {

    private static class Entity {
        private final String id;
        private final String groupId;
        private final float quota;

        public Entity(String id, String groupId, float quota) {
            this.id = id;
            this.groupId = groupId;
            this.quota = quota;
        }

        @Override
        public String toString() {
            return String.format("%s，%s，%f", groupId, id, quota);
        }
    }

    // cpu core num
    private static final int PARALLEL_SORT_THREAD_NUM = Runtime.getRuntime().availableProcessors() * 2;

    // queue
    private static final BlockingQueue<Entity> queue = new LinkedBlockingQueue<>();
    // read thread pool executor
    private static final ExecutorService readGroupExecutor = Executors.newFixedThreadPool(10);
    // sort thread pool executor
    private static final ExecutorService sortGroupExecutor = Executors.newFixedThreadPool(PARALLEL_SORT_THREAD_NUM);


    private static final AtomicLong readCount = new AtomicLong(0);
    private static final AtomicLong sortCount = new AtomicLong(0);
    // 是否读取完成
    private static volatile boolean done = false;

    private static final Map<String, Entity> treeMap = new TreeMap<>(String::compareTo);

    public static void main(String[] args) throws Exception {
        File dir = new File("/Users/guopanbo-work/IdeaProjects/understanding-jvm/data");
        File[] files = dir.listFiles();
        if(files == null)
            throw new FileNotFoundException("file not found");

        CountDownLatch readLatch = new CountDownLatch(files.length);
        for (File file : files) {
            readGroupExecutor.execute(() -> {
//                System.out.println("read file -> " + file.getAbsolutePath());
                try (LineNumberReader reader = new LineNumberReader(new FileReader(file))) {
                    for(;;) {
                        String s = reader.readLine();
                        if(s == null || "".equals(s))
                            break;
                        queue.add(onRead(s));
                        readCount.incrementAndGet();
                    }
                } catch (Exception e) {
                    e.printStackTrace();
                }
                readLatch.countDown();
            });
        }

        CountDownLatch sortLatch = new CountDownLatch(PARALLEL_SORT_THREAD_NUM);
        for (int i = 0; i < PARALLEL_SORT_THREAD_NUM; i++) {
            sortGroupExecutor.execute(() -> {
                for(;;) {
                    try {
                        Entity entity = queue.poll(10, TimeUnit.MILLISECONDS);
                        if(entity != null) {
                            onMessage(entity);
                            sortCount.incrementAndGet();
                        } else {
                            // 处理数量等于读取数量，并且状态是完成，退出
                            if(sortCount.get() == readCount.get() && done)
                                break;
                        }
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                }
                sortLatch.countDown();
            });
        }

        readLatch.await();
//        System.out.println("read file done, count -> " + readCount.get());
        // 标记读取完成
        done = true;
        readGroupExecutor.shutdown();

        sortLatch.await();
//        System.out.println("sort done, count -> " + sortCount.get());
        sortGroupExecutor.shutdown();

        treeMap.values().forEach(System.out::println);
    }

    /**
     * 数据读取到内存中，触发方法的执行
     * @param line
     */
    private static Entity onRead(String line) {
        if (line == null || "".equals(line))
            return null;
        String[] arr = line.split("，");
        if (arr.length < 3)
            throw new RuntimeException("parse data error");

        try {
            return new Entity(arr[0], arr[1], Float.parseFloat(arr[2]));
        } catch (Exception e) {
            e.printStackTrace();
            throw new RuntimeException("parse data error");
        }
    }

    /**
     * 数据到达时
     * @param entity
     */
    private static void onMessage(Entity entity) {
        synchronized (treeMap) {
            Entity oldEntity = treeMap.get(entity.groupId);
            // 不存在，直接放入map
            if (oldEntity == null) {
                treeMap.put(entity.groupId, entity);
            } else {
                // 如果旧的值比新的值大，则替换
                if (oldEntity.quota > entity.quota)
                    treeMap.put(entity.groupId, entity);
            }
        }
    }
}
